use crate::traits::{SmartModule, SmartModuleConfig, ProcessResult, SmartModuleFactory};
use crate::registry::SmartModuleRegistry;
use fluviox_shared::{Result, FluvioXError, models::DataRecord};
use tracing::{info, error, warn};
use std::collections::HashMap;

/// SmartModule处理引擎
/// 管理多个SmartModule的执行流水线
pub struct SmartModuleEngine {
    modules: Vec<Box<dyn SmartModule>>,
    registry: SmartModuleRegistry,
}

impl SmartModuleEngine {
    pub fn new() -> Self {
        Self {
            modules: Vec::new(),
            registry: SmartModuleRegistry::new(),
        }
    }
    
    /// 添加SmartModule到处理流水线
    pub async fn add_module(&mut self, config: SmartModuleConfig) -> Result<()> {
        info!("添加SmartModule: {}", config.name);
        
        if !config.enabled {
            info!("模块 {} 已禁用，跳过", config.name);
            return Ok(());
        }
        
        // 验证配置
        self.registry.validate_config(&config.name, &serde_json::to_value(&config.parameters)?)?;
        
        // 创建模块实例
        let mut module = self.registry.create_module(&config.name)?;
        
        // 初始化模块
        module.initialize(serde_json::to_value(&config.parameters)?).await?;
        
        self.modules.push(module);
        
        info!("成功添加SmartModule: {}", config.name);
        Ok(())
    }
    
    /// 处理单个记录
    pub async fn process_record(&mut self, mut record: DataRecord) -> Result<Vec<DataRecord>> {
        let mut results = vec![record];
        
        for module in &mut self.modules {
            let mut new_results = Vec::new();
            
            for record in results {
                match module.process(record).await {
                    Ok(process_results) => {
                        for result in process_results {
                            match result {
                                ProcessResult::Keep(r) | ProcessResult::Transform(r) => {
                                    new_results.push(r);
                                },
                                ProcessResult::Split(mut rs) => {
                                    new_results.append(&mut rs);
                                },
                                ProcessResult::Filter => {
                                    // 记录被过滤掉
                                },
                                ProcessResult::Aggregate => {
                                    // 记录被聚合，暂不输出
                                },
                            }
                        }
                    },
                    Err(e) => {
                        error!("SmartModule处理失败: {}", e);
                        return Err(e);
                    }
                }
            }
            
            results = new_results;
            
            // 如果没有记录剩余，提前退出
            if results.is_empty() {
                break;
            }
        }
        
        Ok(results)
    }
    
    /// 批量处理记录
    pub async fn process_batch(&mut self, records: Vec<DataRecord>) -> Result<Vec<DataRecord>> {
        let mut all_results = Vec::new();
        
        for record in records {
            let mut record_results = self.process_record(record).await?;
            all_results.append(&mut record_results);
        }
        
        Ok(all_results)
    }
    
    /// 完成处理（用于聚合模块输出最终结果）
    pub async fn finalize(&mut self) -> Result<Vec<DataRecord>> {
        let mut final_results = Vec::new();
        
        for module in &mut self.modules {
            let mut module_results = module.finalize().await?;
            final_results.append(&mut module_results);
        }
        
        Ok(final_results)
    }
    
    /// 重置所有模块状态
    pub async fn reset(&mut self) -> Result<()> {
        for module in &mut self.modules {
            module.reset().await?;
        }
        Ok(())
    }
    
    /// 获取模块数量
    pub fn module_count(&self) -> usize {
        self.modules.len()
    }
    
    /// 清空所有模块
    pub fn clear_modules(&mut self) {
        self.modules.clear();
    }
}

impl Default for SmartModuleEngine {
    fn default() -> Self {
        Self::new()
    }
}
